package com.xxmw.transaction.service.app.component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxmw.transaction.enumeration.BourseEnum;
import com.xxmw.transaction.model.app.BoursePrice;
import com.xxmw.transaction.model.app.Symbol;
import com.xxmw.transaction.service.app.SymbolService;
import com.xxmw.transaction.service.support.AbstractWebSocketAdapter;
import com.xxmw.transaction.service.support.SymbolSubWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.util.Date;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author: yamoo
 * @Date: 2020-02-26 17:25
 **/
@Slf4j
@Component
public final class BitforexApplication  extends AbstractWebSocketAdapter implements SymbolSubWebSocket {

    private static final String BA_URL = "wss://www.bitforex.com/mkapi/coinGroup1/ws";

    private static final String TYPE = "type";

    private static final String SUBSCRIBE = "subHq";

    private static final String UNSUBSCRIBE = "subHq_cancel";

    private static final String PARAMS = "param";

    private static final String EVENT = "event";

    private static final String EVENT_PARAMS = "depth10";

    /**
     * 心跳调度器
     */
    private final ScheduledExecutorService scheduled = Executors.newSingleThreadScheduledExecutor();

    private long period = 5000;

    @Autowired
    private SymbolService symbolService;

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet(new BitForexWebSocketClient(new URI(BA_URL), new Draft_6455(), null, 5000));
        scheduled.scheduleWithFixedDelay(() -> heartBeat(), period, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public void destroy() throws Exception {
        this.client.close();
        this.scheduled.shutdown();
    }

    @Override
    public void subscribe(Symbol symbol) {
        //订阅事件
        sendMessage(symbol,SUBSCRIBE);
    }

    @Override
    public void unSubscribe(Symbol symbol) {
        //取消订阅
        sendMessage(symbol,UNSUBSCRIBE);
        cleanSymbolPrice(symbol.getName());
    }

    @Override
    public void subscribeList(Set<Symbol> subSet) {
        subSet.forEach( item -> subscribe(item));
    }

    @Override
    public BoursePrice getSymbolPrice (String symbol) {
        return symbol_map.get(symbol);
    }

    @Override
    public void cleanSymbolPrice(String symbol) {
        symbol_map.remove(symbol);
    }

    private void sendMessage(Symbol symbol, String method) {
        JSONObject json = new JSONObject();
        JSONObject paramsJson = new JSONObject();
        json.put(TYPE, method);
        json.put(EVENT, EVENT_PARAMS);
        paramsJson.put("businessType", BourseEnum.BITFOREX.getSymbolFormat().symbolFormat(symbol));
        paramsJson.put("dType", 0);
        json.put(PARAMS, paramsJson);
        log.info("发送给币夫消息:{}",json.toJSONString());
        JSONArray array = new JSONArray();
        array.add(json);
        this.client.send(array.toJSONString());
    }



    private class  BitForexWebSocketClient extends WebSocketClient {

        public BitForexWebSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders,
                                       int connectTimeout) {
			super(serverUri, protocolDraft, httpHeaders, connectTimeout);
		}

        @Override
        public void onOpen(ServerHandshake serverHandshake) {
            log.info("币夫WebSocket接口: {} 链接成功.", this.getURI());
            Set<Symbol> subSet = symbolService.listByBourse(BourseEnum.BITFOREX.getValue());
            if (subSet != null && !subSet.isEmpty()) {
                log.info("币夫订阅开始{}",subSet);
                subscribeList(subSet);
            }
        }

        @Override
        public void onMessage(String s) {

            // log.info("接收到币夫websocket消息：{}", s);

            if (s.equals("pong_p")) {
                return;
            }
            JSONObject object = JSONObject.parseObject(s);
            JSONObject data = object.getJSONObject("data");
            if (data == null || data.isEmpty()) {
                log.info("未处理该消息");
                log.info(s);
                return;
            }
            data.put("updateTime",new Date());
            JSONObject param = object.getJSONObject("param");
            String symbol = BourseEnum.BITFOREX.getSymbolFormat().symbolStandardization(param.getString("businessType"));
            JSONArray asksArray = data.getJSONArray("asks");
            JSONArray bidsArray = data.getJSONArray("bids");
            JSONObject asks = asksArray.getJSONObject(0);
            JSONObject bids = bidsArray.getJSONObject(0);
            //  log.info("币夫获取{}买一价为:{},卖一价为:{}",symbol, bids.getString("price"), asks.getString("price"));
            BoursePrice price = BoursePrice.builder().buyPrice(bids.getBigDecimal("price")).buyNum(bids.getBigDecimal("amount")).sellPrice(asks.getBigDecimal("price")).sellNum(bids.getBigDecimal("amount")).data(data.toJSONString()).updateTime(new Date()).build();
            symbol_map.put(symbol, price);
        }

        @Override
        public void onClose(int i, String s, boolean b) {
            log.info("币夫websocket已关闭{},{},{}",i,s,b);
        }

        @Override
        public void onError(Exception e) {
            log.error("币夫websocket异常",e);
        }
    }
}
